package com.ksyun.alert.utils;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.ksyun.alert.calculate.CalculateTiming;
import com.ksyun.alert.calculate.CalculateImmediate;
import com.ksyun.alert.config.CalculationTimingProperties;
import com.ksyun.alert.constant.RedisStreamConstant;
import com.ksyun.alert.dto.MetricDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * @author Zhang
 * @date 2024-05-01
 */
@Component
@Slf4j
public class AnalyzeUtil {
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private CalculateImmediate calculateImmediate;

    /**
     * 结果解析方法
     * @param records 记录
     */
    public List<List<MetricDTO>> analyzeRecord(List<MapRecord<String, Object, Object>> records) {
        LinkedList<List<MetricDTO>> lists = new LinkedList<>();
        records.forEach(record -> {
            Map<Object, Object> recordValue = record.getValue();
            String metrics = recordValue.get(RedisStreamConstant.WRITE_JSON_KEY).toString();
            List<MetricDTO> metricDTOS = new LinkedList<>();
            JSONArray jsonArray = JSON.parseArray(metrics);
            jsonArray.forEach(obj -> {
                JSONObject jsonObject = (JSONObject) obj;
                String metrics1 = jsonObject.getString("metric");
                JSONObject tags = jsonObject.getJSONObject("tags");
                Map<String, String> hashMap = new HashMap<>();
                tags.keySet().forEach(key -> {
                    hashMap.put(key, tags.get(key).toString());
                });
                Long timestamp = jsonObject.getLongValue("timestamp");
                float value = jsonObject.getFloatValue("value");
                MetricDTO build = MetricDTO.builder()
                        .metric(metrics1)
                        .timestamp(timestamp)
                        .value(value)
                        .tags(hashMap)
                        .build();
                metricDTOS.add(build);
            });
            lists.add(metricDTOS);
        });
        return lists;
    }

    /**
     * 实时计算获取Redis消息方法
     * xreadgroup group timing c1 count 1 block 3000 Streams metrics >
     * @param count 获取几条
     * @param consumerName 消费者名称
     */
    public List<MapRecord<String, Object, Object>> readFromStream(int count, String consumerName) {
        StreamOffset<String> offset = StreamOffset.create(RedisStreamConstant.STREAM_NAME, ReadOffset.lastConsumed());
        Consumer consumer = Consumer.from(RedisStreamConstant.IMMEDIATE_GROUP, consumerName);
        StreamReadOptions readOptions = StreamReadOptions.empty().block(Duration.ofSeconds(RedisStreamConstant.BLOCK_TIME)).count(1);
        return stringRedisTemplate.opsForStream().read(consumer, readOptions, offset);
    }

    /**
     * xack确认方法
     */
    public void redisStreamAckImmediate(String ids) {
        stringRedisTemplate.opsForStream().acknowledge(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.IMMEDIATE_GROUP, ids);
    }

    /**
     * xreadgroup group timing c1 count 1 block 3000 Streams metrics >
     * @param group 消费者组名称
     * @param consumerName 消费者名称
     * @param stream 队列名称
     */
    public void handlePendingList(String group, String consumerName, String stream) {
        //失败重试次数
        int handleCount = 0;
        Consumer consumer = Consumer.from(group, consumerName);
        while (true) {
            //尝试获取一个，看看有没有
            PendingMessages pending = stringRedisTemplate.opsForStream().pending(stream, consumer, Range.unbounded(), 1L);
            if (pending.isEmpty()) {
                //pendingList为空。。。返回
                return;
            }
            //获取pendingList数据进行重试
            StreamOffset<String> offset = StreamOffset.create(stream, ReadOffset.from("0"));
            StreamReadOptions readOptions = StreamReadOptions.empty().count(1L);
            List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(consumer, readOptions, offset);
            //第一条数据，总共一条，因为count = 1
            MapRecord<String, Object, Object> entries = records.get(0);
            //id用于xack确认
            String recordId = entries.getId().toString();
            //解析
            List<List<MetricDTO>> lists = analyzeRecord(records);
            //重试
            while (true) {
                try {
                    calculateImmediate.calculateNow(lists.get(0));
                } catch (Exception e) {
                    log.error(e.getMessage());
                    //失败次数加一
                    handleCount++;
                    if (handleCount > 3) {
                        log.error("metrics计算出错超过三次，将由人工处理：metrics:{}", records);
                        //直接确认，交由人工处理
                        stringRedisTemplate.opsForStream().acknowledge(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.IMMEDIATE_GROUP, recordId);
                        return;
                    }
                    continue;
                }
                break;
            }
            //处理成功，确认
            stringRedisTemplate.opsForStream().acknowledge(stream, group, recordId);
        }
    }
}
